package com.vichat.common.rudp.thread;

import com.vichat.common.rudp.RudpBlockData;
import com.vichat.common.rudp.RudpDataUtil;
import com.vichat.common.rudp.RudpMonitorUtil;
import com.vichat.common.rudp.RudpTransData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.DatagramPacket;

public class RudpReceiveThread implements Runnable {
    private static transient Logger logger = LoggerFactory.getLogger(RudpReceiveThread.class);
    private DatagramPacket packet;

    public RudpReceiveThread(DatagramPacket packet) {
        this.packet = packet;
    }

    public void run() {
        long startIme = 0;
        int msgtype = 1;
        if (RudpMonitorUtil.monitorFlag) {
            startIme = System.currentTimeMillis();
            RudpMonitorUtil.startReceiveThread();
        }
        try {

            byte[] bt = packet.getData();
            RudpBlockData blockData = RudpDataUtil.unmarshalBlockData(bt);

            long step1 = System.currentTimeMillis() - startIme;
            if (blockData.msgType == 0) {

                if (RudpMonitorUtil.monitorFlag) {
                    msgtype = 0;
                    //放入trans监控
                    if (msgtype != 1) {
                        RudpMonitorUtil.startTrans(blockData.getStringTrans(), startIme);
                    }
                }


                try {
                    //发送ack
                    RudpBlockData ackData = new RudpBlockData(blockData);
                    ackData.desAddress = packet.getSocketAddress();
                    RudpDataUtil.sendAckPacket(ackData);
                } catch (Exception e) {
                    logger.error("发送ACK报文失败" + blockData.getBlockKey() + "-" + packet.getSocketAddress().toString(), e);
                }

                long step2 = System.currentTimeMillis() - startIme;

                try {
                    RudpTransData transData = RudpDataUtil.assembleData(blockData);
                    if (transData.blockNum == transData.blockList.size()) {
                        if (RudpMonitorUtil.monitorFlag) {
                            //放入trans监控
                            RudpMonitorUtil.endTrans(transData.getStringTrans(), System.currentTimeMillis());
                        }
                        //发送业务处理线程进行处理
                        transData.indicated = true;
                        transData.remoteAddr = packet.getSocketAddress();

                        //String jsonString = RudpDataUtil.assembelBusiString(transData);
                        //logger.error("报文接受成功:"+jsonString);
                        RudpDataUtil.addBusiQueue(transData);
                    } else {
                        logger.info("多宝处理，总共" + transData.blockNum + ",已收到" + transData.receiveNum + "," + blockData.getStringTrans() + blockData.blockId);
                    }

                } catch (Exception e) {
                    logger.info("事务数据组装失败" + blockData.getBlockKey(), e);
                }
                long step3 = System.currentTimeMillis() - startIme;

                if (step3 > 10) {
                    logger.error("!!!!!!!!处理业务报文超时" + step3 + "," + step2 + "," + step1);
                }
            } else {
                //如果接受到的是ack,则设置接受标志

                if (RudpDataUtil.sendDataCont.containsKey(blockData.getBlockKey())) {
                    RudpBlockData sendData = RudpDataUtil.sendDataCont.get(blockData.getBlockKey());
                    sendData.lifeTime = -1;
                    logger.info("有效的ACK的key" + blockData.getBlockKey());
                } else {
                    logger.error("接受到的ACK无效" + blockData.getStringTrans() + "-" + blockData.getBlockKey() + packet.getAddress().getHostAddress());
                }
            }


        } catch (Exception e) {
            logger.error("数据接收处理失败", e);
        }
        if (RudpMonitorUtil.monitorFlag) {
            long costtime = System.currentTimeMillis() - startIme;
            RudpMonitorUtil.endReceiveThread(costtime, msgtype);
        }
    }
}   
